package test

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 在需求一的基础上，增加每个品类用户session的点击统计
 * TOP10热门品类中每个品类的TOP10活跃的Session
 */
object UserVisiterActionTest03 {
  def main(args: Array[String]): Unit = {
    val spark = new SparkConf().setMaster("local[*]").setAppName("UserVisitAction")
    val sc = new SparkContext(spark)
    sc.setLogLevel("ERROR")
    //读取文件
    val sourceRDD = sc.textFile("data/user_visit_action.txt")
    val top10Ids = top10Category(sourceRDD)

    //过滤原始数据，保留点击和前十品类的id
    val filterActionRDD = sourceRDD.filter(
      action => {
        val datas = action.split("_")
        if (datas(6) != "-1") {
          top10Ids.contains(datas(6))
        } else {
          false
        }
      }
    )
    //将品类id和session id进行点击量的统计
    val reduceRDD = filterActionRDD.map(
      action => {
        val datas = action.split("_")
        ((datas(6), datas(2)), 1)
      }
    ).reduceByKey(_ + _)

    //将统计结果转换
    //((品类ID,sessionId),sum) ==>  (品类ID,(sessionId,sum))
    val mapRDD = reduceRDD.map({
      case ((cid, sid), sum) => {
        (cid, (sid, sum))
      }
    })
    //相同的品类进行分组
    val groupRDD = mapRDD.groupByKey()

    //将分组后的数据排序，取前十
    groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    ).collect()
      .foreach(println(_))
    sc.stop()
  }

  def top10Category(sourceRDD: RDD[String]) = {
    //统计品类的点击数量(商品品类，点击总量)
    val clickRDD = sourceRDD.map(item => (item.split("_")(6), 1))
      .filter(_._1 != "-1")
      .reduceByKey(_ + _)
    //统计品类的下单数（商品品类，下单总数）
    val orderRDD = sourceRDD.map(_.split("_")(8))
      .filter(_ != "null")
      .flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey(_ + _)
    //统计品类的支付数（商品品类，支付总数）
    val payRDD = sourceRDD.map(_.split("_")(10))
      .filter(_ != "null")
      .flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey(_ + _)
    //组合
    /**
     * 求出三个的总数
     * 此时的三个RDD的数据形式是
     * (商品品类，总点击量)
     * (商品品类，总订单量)
     * (商品品类，总支付量)
     * 需要的形式(商品品类，(点击量，订单量，支付量))
     * 我们可以进行如下的转换
     * (商品品类，点击量)  ==>  (商品品类，（点击量，0，0）)
     * (商品品类，订单量)  ==>  (商品品类，（0，订单量，0）)
     * (商品品类，支付量)  ==>  (商品品类，（0，0，支付量）)
     * 然后对数据来进行集合操作即可
     */
    //第一步：将数据转换变形
    val rdd1 = clickRDD.map {
      case (cid, cnt) => {
        (cid, (cnt, 0, 0))
      }
    }
    val rdd2 = orderRDD.map {
      case (cid, cnt) => {
        (cid, (0, cnt, 0))
      }
    }
    val rdd3 = payRDD.map {
      case (cid, cnt) => {
        (cid, (0, 0, cnt))
      }
    }
    //进行聚合操作
    val result = rdd1.union(rdd2).union(rdd3).reduceByKey((v1, v2) => (v1._1 + v2._1, v1._2 + v2._2, v1._3 + v2._3))
    result.sortBy(_._2, false)
      .take(10)
      .map(_._1)
  }

}
